package com.example.service;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.core.io.ClassPathResource;
import org.springframework.core.io.Resource;
import org.springframework.data.domain.Range;
import org.springframework.data.redis.connection.stream.MapRecord;
import org.springframework.data.redis.connection.stream.RecordId;
import org.springframework.data.redis.connection.stream.StreamRecords;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.core.script.DefaultRedisScript;
import org.springframework.scripting.support.ResourceScriptSource;
import org.springframework.stereotype.Service;

import java.util.*;
import java.util.concurrent.TimeUnit;

@Service
public class RedisService {

    private static final Logger logger = LoggerFactory.getLogger(RedisService.class);

    // Redis 键前缀
    private static final String PREFIX = "testRedis:";

    // 定义常量键
    public static final String KEY_QUEUE = PREFIX + "queue";
    public static final String KEY_LEADERBOARD = PREFIX + "leaderboard";
    public static final String KEY_ONLINE_USERS = PREFIX + "onlineUsers";
    public static final String KEY_USER_LOCATIONS = PREFIX + "userLocations";
    public static final String KEY_HYPERLOGLOG = PREFIX + "hyperLogLog";
    public static final String KEY_BITMAP = PREFIX + "bitmap";
    public static final String KEY_STRING_VALUE = PREFIX + "stringValue";
    public static final String KEY_HASH = PREFIX + "hash";
    public static final String KEY_STREAM = PREFIX + "stream";

    private final RedisTemplate<String, Object> redisTemplate;
    private final StringRedisTemplate stringRedisTemplate;

    @Autowired
    public RedisService(RedisTemplate<String, Object> redisTemplate, StringRedisTemplate stringRedisTemplate) {
        this.redisTemplate = redisTemplate;
        this.stringRedisTemplate = stringRedisTemplate;
    }

    // ================ 基本数据类型操作（Key-Value） ==================

    /**
     * 设置字符串值
     *
     * @param value 要设置的值
     */
    public void setValue(String value) {
        logger.info("设置字符串键值: key={}, value={}", KEY_STRING_VALUE, value);
        stringRedisTemplate.opsForValue().set(KEY_STRING_VALUE, value);
    }

    /**
     * 获取字符串值
     *
     * @return 获取的值
     */
    public String getValue() {
        String value = stringRedisTemplate.opsForValue().get(KEY_STRING_VALUE);
        logger.info("获取字符串键值: key={}, value={}", KEY_STRING_VALUE, value);
        return value;
    }

    // ================ 哈希（Hash） ==================

    /**
     * 设置哈希值
     *
     * @param key   哈希键
     * @param value 哈希值
     */
    public void setHashValue(String key, Object value) {
        logger.info("设置哈希 {} 中的键值对: key={}, value={}", KEY_HASH, key, value);
        redisTemplate.opsForHash().put(KEY_HASH, key, value);
    }

    /**
     * 获取哈希值
     *
     * @param key 哈希键
     * @return 获取的哈希值
     */
    public Object getHashValue(String key) {
        Object value = redisTemplate.opsForHash().get(KEY_HASH, key);
        logger.info("获取哈希 {} 中的键: {}, 值: {}", KEY_HASH, key, value);
        return value;
    }

    /**
     * 获取哈希中的所有键值对
     *
     * @return 所有键值对
     */
    public Map<Object, Object> getAllHashValues() {
        Map<Object, Object> entries = redisTemplate.opsForHash().entries(KEY_HASH);
        logger.info("获取哈希 {} 中的所有键值对: {}", KEY_HASH, entries);
        return entries;
    }

    /**
     * 删除哈希键
     *
     * @param key 要删除的哈希键
     */
    public void deleteHashKey(String key) {
        logger.info("删除哈希 {} 中的键: {}", KEY_HASH, key);
        redisTemplate.opsForHash().delete(KEY_HASH, key);
    }

    // ================ 列表（List） ==================

    /**
     * 添加值到队列
     *
     * @param value 要添加的值
     */
    public void addToQueue(String value) {
        logger.info("向队列 {} 添加值: {}", KEY_QUEUE, value);
        stringRedisTemplate.opsForList().rightPush(KEY_QUEUE, value);
    }

    /**
     * 从队列中弹出值
     *
     * @return 弹出的值
     */
    public String popFromQueue() {
        String value = stringRedisTemplate.opsForList().leftPop(KEY_QUEUE);
        logger.info("从队列 {} 中弹出值: {}", KEY_QUEUE, value);
        return value;
    }

    // ================ 集合（Set） ==================

    /**
     * 添加在线用户
     *
     * @param userId     用户ID
     * @param expiration 过期时间
     * @param timeUnit   时间单位
     */
    public void addOnlineUser(String userId, long expiration, TimeUnit timeUnit) {
        logger.info("添加在线用户 {} 到集合 {}", userId, KEY_ONLINE_USERS);
        stringRedisTemplate.opsForSet().add(KEY_ONLINE_USERS, userId);
        stringRedisTemplate.expire(KEY_ONLINE_USERS, expiration, timeUnit);
    }

    /**
     * 获取在线用户数量
     *
     * @return 在线用户数量
     */
    public Long getOnlineUsersCount() {
        Long count = stringRedisTemplate.opsForSet().size(KEY_ONLINE_USERS);
        logger.info("获取集合 {} 的在线用户数: {}", KEY_ONLINE_USERS, count);
        return count;
    }

    // ================ 排行榜（Sorted Set） ==================

    /**
     * 添加玩家到排行榜
     *
     * @param player 玩家名称
     * @param score  玩家分数
     */
    public void addToLeaderboard(String player, double score) {
        logger.info("向排行榜 {} 添加玩家: {}, 分数: {}", KEY_LEADERBOARD, player, score);
        stringRedisTemplate.opsForZSet().add(KEY_LEADERBOARD, player, score);
    }

    /**
     * 获取排行榜的前N名
     *
     * @param topN 前N名
     * @return 排行榜的玩家集合
     */
    public Set<String> getTopFromLeaderboard(int topN) {
        Set<String> topPlayers = stringRedisTemplate.opsForZSet().reverseRange(KEY_LEADERBOARD, 0, topN - 1);
        logger.info("获取排行榜 {} 的前 {} 名玩家: {}", KEY_LEADERBOARD, topN, topPlayers);
        return topPlayers;
    }

    // ================ 地理位置（Geo） ==================

    /**
     * 保存用户位置信息
     *
     * @param user      用户ID
     * @param latitude  纬度
     * @param longitude 经度
     */
    public void saveUserLocation(String user, double latitude, double longitude) {
        logger.info("保存用户 {} 的位置信息: 纬度 {}, 经度 {}", user, latitude, longitude);
        stringRedisTemplate.opsForGeo().add(KEY_USER_LOCATIONS, new org.springframework.data.geo.Point(longitude, latitude), user);
    }

    /**
     * 获取两个用户之间的距离
     *
     * @param user1 用户1 ID
     * @param user2 用户2 ID
     * @return 距离（米）
     */
    public Double getUserDistance(String user1, String user2) {
        Double distance = stringRedisTemplate.opsForGeo().distance(KEY_USER_LOCATIONS, user1, user2).getValue();
        logger.info("计算用户 {} 和 用户 {} 之间的距离: {} 米", user1, user2, distance);
        return distance;
    }

    // ================ 基数统计（HyperLogLog） ==================

    /**
     * 添加值到 HyperLogLog
     *
     * @param value 要添加的值
     */
    public void addToHyperLogLog(String value) {
        logger.info("向 HyperLogLog {} 中添加值: {}", KEY_HYPERLOGLOG, value);
        stringRedisTemplate.opsForHyperLogLog().add(KEY_HYPERLOGLOG, value);
    }

    /**
     * 获取 HyperLogLog 基数统计
     *
     * @return 基数统计值
     */
    public Long getHyperLogLogCount() {
        Long count = stringRedisTemplate.opsForHyperLogLog().size(KEY_HYPERLOGLOG);
        logger.info("获取 HyperLogLog {} 的基数统计: {}", KEY_HYPERLOGLOG, count);
        return count;
    }

    // ================ 位图（Bitmap） ==================

    /**
     * 设置位图中的位
     *
     * @param offset 位偏移量
     * @param value  位值
     */
    public void setBit(long offset, boolean value) {
        logger.info("设置 Bitmap {} 中的位 {} 为: {}", KEY_BITMAP, offset, value);
        stringRedisTemplate.opsForValue().setBit(KEY_BITMAP, offset, value);
    }

    /**
     * 获取位图中的位
     *
     * @param offset 位偏移量
     * @return 位值
     */
    public Boolean getBit(long offset) {
        Boolean bitValue = stringRedisTemplate.opsForValue().getBit(KEY_BITMAP, offset);
        logger.info("获取 Bitmap {} 中的位 {} 的值: {}", KEY_BITMAP, offset, bitValue);
        return bitValue;
    }

    // ================ 流（Stream） ==================

    /**
     * 向 Stream 中添加数据
     *
     * @param field 字段名称
     * @param value 字段值
     * @return 记录的 ID
     */
    public String addToStream(String field, String value) {
        logger.info("向 Stream {} 中添加数据: {} -> {}", KEY_STREAM, field, value);

        // 使用 MapRecord 构建 Record
        Map<String, String> data = new HashMap<>();
        data.put(field, value);

        MapRecord<String, String, String> record = StreamRecords.newRecord()
                .in(KEY_STREAM)
                .ofMap(data);

        RecordId recordId = redisTemplate.opsForStream().add(record);
        logger.info("Stream 记录ID: {}", recordId);

        return recordId.getValue();
    }

    /**
     * 读取 Stream 中的数据
     *
     * @param startId 起始ID
     * @param endId   结束ID
     * @return 读取的记录列表
     */
    public List<MapRecord<String, Object, Object>> readFromStream(String startId, String endId) {
        logger.info("读取 Stream {} 中的记录, 从ID: {} 到ID: {}", KEY_STREAM, startId, endId);

        // 构建 Range 对象
        Range<String> range = Range.closed(startId, endId);

        List<MapRecord<String, Object, Object>> records = redisTemplate.opsForStream().range(KEY_STREAM, range);
        return records;
    }

    /**
     * 更新排行榜中的玩家分数，累加分数，并返回更新后的实际积分（不包含时间戳的微调部分）。
     *
     * @param uid            玩家 UID
     * @param incrementScore 需要增加的分数
     * @param isEarlyHigher  是否早到的分数更高
     * @return 更新后的实际积分
     */
    public Long updateLeaderboardScore(String uid, double incrementScore, boolean isEarlyHigher) {
        // 设置排行榜名称
        String leaderboardName = KEY_LEADERBOARD;

        // 获取当前时间戳
        long timestamp = System.currentTimeMillis();

        logger.info("更新排行榜 {} 中玩家 {} 的分数: {}, 时间戳: {}, 早到分数更高: {}", leaderboardName, uid, incrementScore, timestamp, isEarlyHigher);

        // 加载 Lua 脚本
        Resource scriptResource = new ClassPathResource("scripts/update_leaderboard_with_timestamp.lua");
        DefaultRedisScript<Long> redisScript = new DefaultRedisScript<>();
        redisScript.setScriptSource(new ResourceScriptSource(scriptResource));
        redisScript.setResultType(Long.class);

        // 执行 Lua 脚本，并返回更新后的实际积分
        Long newScore = redisTemplate.execute(
                redisScript,
                Collections.singletonList(leaderboardName),
                uid,
                String.valueOf(incrementScore),
                String.valueOf(timestamp),
                String.valueOf(isEarlyHigher)
        );

        logger.info("Lua 脚本执行结果，更新后的实际积分: {}", newScore);
        return newScore;
    }
}
